package com.cellulam.framework.mq.listener;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;
import com.cellulam.framework.mq.ActionResult;
import lombok.extern.slf4j.Slf4j;

import java.util.Properties;

@Slf4j
public abstract class MqOrderedMessageListener
        extends BaseAbstractMessageListener<OrderConsumer> {

    protected MqOrderedMessageListener() {
        this(MqConsumeMessageMode.CLUSTERING);
    }

    protected MqOrderedMessageListener(MqConsumeMessageMode messageMode) {
        super(messageMode);
    }

    @Override
    protected final OrderConsumer createConsumer(Properties properties) {
        return ONSFactory.createOrderedConsumer(properties);
    }

    @Override
    public final void subscribe(String groupId, String topic, String subExpression) {
        log.info("订阅, groupId: {}, topic: {}, subExpression: {}",
                groupId, topic, subExpression);

        super.getConsumer().subscribe(topic, subExpression, ((message, context) -> {
            ActionResult result = super.consumeMessage(message);
            return this.convertToOrderAction(result);
        }));
    }

    private OrderAction convertToOrderAction(ActionResult result) {
        switch (result) {
            case SUCCESS:
                return OrderAction.Success;
            default:
                return OrderAction.Suspend;
        }
    }
}
